package bigdata.hermesfuxi.eagle.etl.jobs; // 定义包名，表示这是ETL作业相关的类

import bigdata.hermesfuxi.eagle.etl.bean.DataLogBean; // 导入数据日志Bean类，用于封装用户行为数据
import bigdata.hermesfuxi.eagle.etl.functions.LocationAsyncAMapFunction; // 导入异步地理位置查询函数，用于根据经纬度获取地址信息
import bigdata.hermesfuxi.eagle.etl.utils.FlinkUtils; // 导入Flink工具类，提供Flink环境配置和Kafka数据源
import bigdata.hermesfuxi.eagle.etl.utils.MyKafkaDeserializationSchema; // 导入自定义Kafka反序列化器
import ch.hsr.geohash.GeoHash; // 导入GeoHash工具类，用于将经纬度转换为GeoHash编码
import com.alibaba.fastjson.JSON; // 导入FastJSON库，用于JSON数据解析
import org.apache.flink.api.java.tuple.Tuple2; // 导入Flink二元组类，用于存储Kafka消息的key和value
import org.apache.flink.connector.jdbc.JdbcConnectionOptions; // 导入JDBC连接配置类
import org.apache.flink.connector.jdbc.JdbcSink; // 导入JDBC Sink，用于将数据写入数据库
import org.apache.flink.streaming.api.datastream.AsyncDataStream; // 导入异步数据流处理类
import org.apache.flink.streaming.api.datastream.DataStream; // 导入数据流基类
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; // 导入单输出流操作符
import org.apache.flink.streaming.api.functions.ProcessFunction; // 导入处理函数基类
import org.apache.flink.util.Collector; // 导入数据收集器

import java.time.LocalDateTime; // 导入本地日期时间类
import java.time.ZoneOffset; // 导入时区偏移量类
import java.time.format.DateTimeFormatter; // 导入日期时间格式化器
import java.util.concurrent.TimeUnit; // 导入时间单位枚举

import static bigdata.hermesfuxi.eagle.etl.constant.Constants.GEO_HASH_LENGTH; // 导入GeoHash长度常量


/**
 * 将数据存储在 ClickHouse 中
 * @author Hermesfuxi-PC
 */
public class DataToClickHouse {
    public static void main(String[] args) throws Exception { // 主方法入口
        // 从Kafka获取数据源，返回包含消息key和value的二元组数据流
        DataStream<Tuple2<String, String>> kafkaSourceV2 = FlinkUtils.getKafkaSourceV2(args, MyKafkaDeserializationSchema.class);
//        kafkaSourceV2.print(); // 注释掉的打印语句，用于调试

        //解析数据
        // 使用ProcessFunction处理Kafka数据，将JSON字符串转换为DataLogBean对象
        SingleOutputStreamOperator<DataLogBean> beanStream = kafkaSourceV2.process(new ProcessFunction<Tuple2<String, String>, DataLogBean>() {
            @Override
            public void processElement(Tuple2<String, String> value, Context ctx, Collector<DataLogBean> out) throws Exception {
                try { // 开始异常处理
                    // 将JSON字符串解析为DataLogBean对象
                    DataLogBean dataLogBean = JSON.parseObject(value.f1, DataLogBean.class);
                    if (dataLogBean != null) { // 检查解析结果是否为空
                        dataLogBean.setGid(value.f0); // 设置全局ID（来自Kafka消息的key）
                        // 检查经纬度是否都存在
                        if (dataLogBean.getLatitude() != null && dataLogBean.getLongitude() != null) {
                            Double latitude = dataLogBean.getLatitude(); // 获取纬度
                            Double longitude = dataLogBean.getLongitude(); // 获取经度
                            // 将经纬度转换为GeoHash编码，用于地理位置索引
                            String geoHashCode = GeoHash.geoHashStringWithCharacterPrecision(latitude, longitude, GEO_HASH_LENGTH);
                            dataLogBean.setGeoHashCode(geoHashCode); // 设置GeoHash编码
                        }

                        // 检查时间戳是否存在
                        if(dataLogBean.getTimestamp()!= null){
                            // 将毫秒时间戳转换为本地日期时间（东八区）
                            LocalDateTime localDateTime = LocalDateTime.ofEpochSecond(dataLogBean.getTimestamp() / 1000, 0, ZoneOffset.ofHours(8));
                            String date = localDateTime.toLocalDate().toString(); // 提取日期字符串
                            dataLogBean.setDate(date); // 设置日期
                            dataLogBean.setHour(localDateTime.getHour()); // 设置小时
                        }
                        out.collect(dataLogBean); // 输出处理后的数据
                    }
                } catch (Exception e) { // 捕获所有异常
                    e.printStackTrace(); // 打印异常堆栈信息
                }
            }
        
        }
        );
//        beanStream.print(); // 注释掉的打印语句，用于调试

        int capacity = 10; // 设置异步请求队列容量为10
        // 使用异步数据流处理地理位置查询，保持数据顺序
        SingleOutputStreamOperator<DataLogBean> result = AsyncDataStream.orderedWait(
                beanStream, //输入的数据流
                new LocationAsyncAMapFunction(capacity), //异步查询的Function实例
                300000, //超时时间（300秒）
                TimeUnit.MILLISECONDS, //时间单位（毫秒）
                capacity).setParallelism(4);//异步请求队列最大的数量，不传该参数默认值为100

        result.print(); // 打印处理结果，用于调试



        // 添加JDBC Sink，将数据写入ClickHouse数据库
        result.addSink(JdbcSink.sink(
                // SQL插入语句，包含所有字段
                "insert into doit.tb_user_event (gid,carrier,deviceId,deviceType,eventId,isNew,country,province,city,district,netType,osName,osVersion,releaseChannel,resolution,sessionId,timestamp,date,hour,processTime) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
                (ps, bean) -> { // Lambda表达式，设置SQL参数
                    ps.setString(1, bean.getGid()); // 设置全局ID
                    ps.setString(2, bean.getCarrier()); // 设置运营商
                    ps.setString(3, bean.getDeviceId()); // 设置设备ID
                    ps.setString(4, bean.getDeviceType()); // 设置设备类型
                    ps.setString(5, bean.getEventId()); // 设置事件ID
                    ps.setInt(6, bean.getIsNew()); // 设置是否新用户
                    ps.setString(7, bean.getCountry()); // 设置国家
                    ps.setString(8, bean.getProvince()); // 设置省份
                    ps.setString(9, bean.getCity()); // 设置城市
                    ps.setString(10, bean.getDistrict()); // 设置区县
                    ps.setString(11, bean.getNetType()); // 设置网络类型
                    ps.setString(12, bean.getOsName()); // 设置操作系统名称
                    ps.setString(13, bean.getOsVersion()); // 设置操作系统版本
                    ps.setString(14, bean.getReleaseChannel()); // 设置发布渠道
                    ps.setString(15, bean.getResolution()); // 设置屏幕分辨率
                    ps.setString(16, bean.getSessionId()); // 设置会话ID
                    ps.setLong(17, bean.getTimestamp()); // 设置时间戳
                    ps.setString(18, bean.getDate()); // 设置日期
                    ps.setInt(19, bean.getHour()); // 设置小时
                    ps.setString(20, LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))); // 设置处理时间
                },
                // 配置ClickHouse数据库连接
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:clickhouse://hadoop-master:8123/doit") // 设置数据库连接URL
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver") // 设置ClickHouse驱动
                        .build())); // 构建连接配置

        FlinkUtils.env.execute(); // 执行Flink作业
    }
}

/* clickhouse 建表语句：

create table tb_user_event(
gid String,
carrier String,
deviceId String,
deviceType String,
eventId String,
isNew UInt8,
country String,
province String,
city String,
district String,
netType String,
osName String,
osVersion String,
releaseChannel String,
resolution String,
sessionId String,
timestamp UInt16,
date String,
hour UInt8,
processTime DateTime comment '插入到数据库时的系统时间'
)engine = ReplacingMergeTree(processTime)
partition by (date, hour)
order by gid;

 */
